package com.fbs.job.core.server;

import com.fbs.job.core.biz.ExecutorBiz;
import com.fbs.job.core.biz.impl.ExecutorBizImpl;
import com.fbs.job.core.biz.model.LogParam;
import com.fbs.job.core.biz.model.TriggerParam;
import com.fbs.job.core.model.ResultT;
import com.fbs.job.core.thread.ExecutorRegistryThread;
import com.fbs.job.core.util.FbsJobRemotingUtil;
import com.fbs.job.core.util.JsonTool;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;

public class EmbedServer {
    private static final Logger logger = LoggerFactory.getLogger(EmbedServer.class);
    private ExecutorBiz executorBiz;
    private Thread thread;
    public void start(final String address, final int port, final String appName, final String accessToken){
        executorBiz = new ExecutorBizImpl();
        thread = new Thread(new Runnable() {
            @Override
            public void run() {
                //定义线程池
                ThreadPoolExecutor bizThreadPool = new ThreadPoolExecutor(
                        0, 200, 60L, TimeUnit.SECONDS,
                        new LinkedBlockingDeque<>(2000),
                        new ThreadFactory() {
                            @Override
                            public Thread newThread(Runnable r) {
                                return new Thread(r, "fbs-job, EmbedServer bizThreadPool-" + r.hashCode());
                            }
                        },
                        new RejectedExecutionHandler() {
                            @Override
                            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                                throw new RuntimeException("fbs-job, EmbedServer bizThreadPool is EXHAUSTED!");
                            }
                        }
                );
                EventLoopGroup bossGroup = new NioEventLoopGroup();
                EventLoopGroup workerGroup = new NioEventLoopGroup();
                try{
                    // start server
                    ServerBootstrap bootstrap = new ServerBootstrap();
                    bootstrap.group(bossGroup,workerGroup)//1
                         .channel(NioServerSocketChannel.class)
                         .childHandler(new ChannelInitializer<SocketChannel>() {
                                @Override
                                protected void initChannel(SocketChannel channel) throws Exception {
                                     channel.pipeline()
                                            .addLast(new IdleStateHandler(0,0,30*3,TimeUnit.SECONDS))
                                            .addLast(new HttpServerCodec())
                                            .addLast(new HttpObjectAggregator(5*1025*1024))
                                            .addLast(new EmbedHttpServerHandler(executorBiz,accessToken,bizThreadPool))
                                    ;
                                }
                    }).childOption(ChannelOption.SO_KEEPALIVE,true);


                    // bind
                    ChannelFuture future = bootstrap.bind(port).sync();
                    logger.info(">>>>>>>>>>> fbs-job remoting server start success, nettype = {}, port = {}", EmbedServer.class, port);
                    // start registry
                    // 客户端注册到调度中心
                    startRegistry(appName, address);

                    // wait util stop
                    future.channel().closeFuture().sync();
                } catch (InterruptedException e) {
                    logger.info(">>>>>>>>>>> fbs-job remoting server stop.");
                } catch (Exception e) {
                    logger.error(">>>>>>>>>>> fbs-job remoting server error.", e);
                } finally {
                    // stop
                    try {
                        workerGroup.shutdownGracefully();
                        bossGroup.shutdownGracefully();
                    } catch (Exception e) {
                        logger.error(e.getMessage(), e);
                    }
                }


            }
        });
        thread.setDaemon(true);    // daemon, service jvm, user thread leave >>> daemon leave >>> jvm leave
        thread.start();
    }



    // ---------------------- registry ----------------------

    public void startRegistry(final String appName, final String address) {
        // start registry
        ExecutorRegistryThread.getInstance().start(appName, address);
    }

    public void stop(){
        // destroy server thread
        if (thread != null && thread.isAlive()) {
            thread.interrupt();
        }

        // stop registry
        stopRegistry();
        logger.info(">>>>>>>>>>> fbs-job remoting server destroy success.");
    }
    public void stopRegistry() {
        // stop registry
        ExecutorRegistryThread.getInstance().toStop();
    }

    public static class EmbedHttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
        private static final Logger logger = LoggerFactory.getLogger(EmbedHttpServerHandler.class);
        private ExecutorBiz executorBiz;
        private String accessToken;
        private ThreadPoolExecutor bizThreadPool;
        public EmbedHttpServerHandler(ExecutorBiz executorBiz,String accessToken,ThreadPoolExecutor bizThreadPool){
             this.executorBiz = executorBiz;
             this.accessToken = accessToken;
             this.bizThreadPool = bizThreadPool;
         }
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception {
             // 执行admin发送的任务命令
            String requestData = msg.content().toString(CharsetUtil.UTF_8);
            String uri = msg.uri();
            HttpMethod httpMethod = msg.method();

            boolean keepAlive = HttpUtil.isKeepAlive(msg);
            String accessTokenReq = msg.headers().get(FbsJobRemotingUtil.FBS_JOB_ACCESS_TOKEN);
            bizThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    // do invoke
                    Object responseObj = process(httpMethod, uri, requestData, accessTokenReq);

                    // to json
                    String responseJson = JsonTool.toJson(responseObj);

                    // write response
                    writeResponse(ctx, keepAlive, responseJson);
                }
            });

        }
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.flush();
        }
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            logger.error(">>>>>>>>>>> fbs-job provider netty_http server caught exception", cause);
            ctx.close();
        }
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                ctx.channel().close();      // beat 3N, close if idle
                logger.debug(">>>>>>>>>>> fbs-job provider netty_http server close an idle channel.");
            } else {
                super.userEventTriggered(ctx, evt);
            }
        }
        /**
         * write response
         */
        private void writeResponse(ChannelHandlerContext ctx, boolean keepAlive, String responseJson) {
            // write response
            FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.copiedBuffer(responseJson, CharsetUtil.UTF_8));   //  Unpooled.wrappedBuffer(responseJson)
            response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html;charset=UTF-8");       // HttpHeaderValues.TEXT_PLAIN.toString()
            response.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
            if (keepAlive) {
                response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
            }
            ctx.writeAndFlush(response);
        }
        private Object process(HttpMethod httpMethod, String uri, String requestData, String accessTokenReq){
            // valid
            if (HttpMethod.POST != httpMethod) {
                return new ResultT<String>(ResultT.FAIL_CODE, "invalid request, HttpMethod not support.");
            }
            if (uri == null || uri.trim().length() == 0) {
                return new ResultT<String>(ResultT.FAIL_CODE, "invalid request, uri-mapping empty.");
            }
            if (accessToken != null
                    && accessToken.trim().length() > 0
                    && !accessToken.equals(accessTokenReq)) {
                return new ResultT<String>(ResultT.FAIL_CODE, "The access token is wrong.");
            }
            // services mapping
            try {
                switch (uri) {
                    case "/beat":
                        return executorBiz.beat();
                    case "/run":
                        TriggerParam triggerParam = JsonTool.fromJson(requestData, TriggerParam.class);
                        return executorBiz.run(triggerParam);
                    case "/log":
                        LogParam logParam = JsonTool.fromJson(requestData, LogParam.class);
                        return executorBiz.log(logParam);
                    default:
                        return new ResultT<String>(ResultT.FAIL_CODE, "invalid request, uri-mapping(" + uri + ") not found.");
                }
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
                return new ResultT<String>(ResultT.FAIL_CODE, "request error:" + e.getMessage());
            }

        }
    }




}


